package service

import (
	"errors"
	"strings"
	"time"
	"vrcm/src/vrcm/app/admin/models"
	"vrcm/src/vrcm/app/admin/models/live_schedule"
	dto "vrcm/src/vrcm/app/admin/service/dto/live_schedule"
	StreamHubDto "vrcm/src/vrcm/app/admin/service/dto/stream_hub"
	liveService "vrcm/src/vrcm/app/admin/service/live"
	streamHubService "vrcm/src/vrcm/app/admin/service/stream_hub"
	cDto "vrcm/src/vrcm/common/dto"
	"vrcm/src/vrcm/common/service"
	ext "vrcm/src/vrcm/envconfig"
)

type RegisterEquipService struct {
	service.Service
}

func (e *RegisterEquipService) InsertEquip(c *dto.SourceBasicInsertReq) error {

	tx := e.Orm.Begin()
	defer func() {
		if r := recover(); r != nil {
			e.Log.Errorf("SourceBasicService Insert error:%s \r\n", r)
			tx.Rollback()
			//return
		}
	}()
	if err := tx.Error; err != nil {
		e.Log.Errorf("SourceBasicService Insert error:%s \r\n", err)
		return err
	}

	c.Kind = 4
	c.StreamType = 1
	c.ActiveItem = 1
	c.SwitchType = 2
	c.Area = 1
	c.IsCustom = 2
	c.EnableAutoSwitch = 2
	c.ProtocolType = 1
	c.SourceType = 2
	c.Details = make([]dto.SourceDetailDto, 0, 3)
	for i := 0; i < 4; i++ {
		detail := dto.SourceDetailDto{
			SourceUrl: "",
			UrlType:   i + 1,
			IsEnable:  2,
		}
		if i == 0 {
			detail.IsEnable = 1
		}
		c.Details = append(c.Details, detail)
	}
	liveService := &liveService.LiveBasic{}
	streamHubService := &streamHubService.StreamHubService{}
	streamHubService.Log = e.Log
	liveService.Log = e.Log

	var sourceBasic live_schedule.SourceBasic
	err := tx.Model(&live_schedule.SourceBasic{}).Where("equip_no=?", c.EquipNo).Find(&sourceBasic).Error

	if err != nil {
		e.Log.Errorf("get SourceBasic error:%s \r\n", err)
		tx.Rollback()
		return err
	}
	if sourceBasic.Id > 0 {
		return errors.New("Duplicate entry")
	}
	//判断主备辅信源是否存在相同
	m := make(map[string]struct{})
	for _, v := range c.Details {
		if len(v.SourceUrl) <= 0 {
			continue
		}
		_, ok := m[v.SourceUrl]
		if ok {
			return errors.New("流地址不能重复")
		}
		m[v.SourceUrl] = struct{}{}
	}

	//now := time.Now().Unix()
	var data live_schedule.SourceBasic
	c.Generate(&data)
	data.EquipState = 2
	err = tx.Create(&data).Error
	if err != nil {
		e.Log.Errorf("SourceBasicService Insert error:%s \r\n", err)
		tx.Rollback()
		return err
	}

	//组装信源详情
	sourceDetails := GenerateSourceDetailParam(c)

	sources := make([]StreamHubDto.Source, 0, 3)
	//推流+公网 需要调用腾讯直播获取拉流地址
	if c.StreamType == 1 && c.Area == 1 {
		for k, v := range sourceDetails {
			//如果禁用不用调用接口获取udp
			if v.IsEnable == 2 {
				continue
			}
			//否则先去全球直播获取对应的拉流地址
			streamReq := GenerateInsertLiveParam(c)
			res, err := liveService.CreateLiveStream(streamReq)
			if err != nil {
				e.Log.Errorf("CreateLiveStream error:%s \r\n", err)
				tx.Rollback()
				return err
			}
			//更新流地址
			//streamType := protocols[c.ProtocolType]
			pushes := strings.Split(res.PushUrl, models.Delimiter)

			*(&sourceDetails[k].PullUrl) = res.RtmpUrl
			*(&sourceDetails[k].LiveId) = res.LiveId
			pushUrl := ""
			if len(pushes) > 0 {
				for _, v := range pushes {
					if strings.Contains(v, "rtmp") || strings.Contains(v, "srt") {
						pushUrl += v + models.Delimiter
					}
				}
			}
			*(&sourceDetails[k].PushUrl) = strings.Trim(pushUrl, models.Delimiter)
			*(&sourceDetails[k].SourceUrl) = res.RtmpUrl
		}
	}

	now := time.Now().Unix()
	//组建控制协转接口输入参数
	for _, v := range sourceDetails {
		if v.IsEnable == 2 {
			continue
		}
		source := StreamHubDto.Source{}
		source.Url = v.SourceUrl
		if len(v.PushUrl) > 0 {
			source.Url = v.PullUrl
		}
		sources = append(sources, source)
	}

	//最后调用控制协转接口
	req := StreamHubDto.StreamHubReq{}
	req.Name = c.SourceName
	req.Type = "live"
	req.TenantId = c.TenantId
	req.Description = ""
	req.Input = StreamHubDto.Input{
		Format: protocols[c.ProtocolType],
		Active: 0,
		Policy: StreamHubDto.Policy{
			Mode: "manual",
		},
		Sources: sources,
	}

	//推流+内网 类型为push 否则pull
	if c.StreamType == 1 && c.Area == 2 {
		req.Input.Type = streamTypes[1]
	} else {
		req.Input.Type = streamTypes[2]
	}

	req.CallbackAddress = ext.ExtConfig.ServiceConfig.VrcmServiceUrl + "/api/v1/source-basic/source-call-back"
	delay := StreamHubDto.Delay{
		Enable:   "no",
		Duration: 10,
	}
	req.Output = StreamHubDto.Output{
		Targets: []StreamHubDto.Target{},
		Delay:   delay,
		Acl:     StreamHubDto.Acl{},
	}

	//最后调用hub接口
	hub, err := streamHubService.CreateStreamHub(&req)
	if err != nil {
		e.Log.Errorf("CreateStreamHub error:%s \r\n", err)
		tx.Rollback()
		return err
	}

	//更新信源详情
	for k, v := range sourceDetails {
		*(&sourceDetails[k].TaskId) = hub.TaskId
		if c.Area == 3 {
			*(&sourceDetails[k].Area) = 1
		}
		*(&sourceDetails[k].SourceId) = data.Id
		if v.IsEnable == 2 {
			continue
		}
		//记录节点id 整合数据入库
		for _, source := range hub.Input.Sources {
			if source.Url == v.SourceUrl && ((source.Num+1) == sourceDetails[k].UrlType) || len(hub.Input.Sources) == 1 {
				*(&sourceDetails[k].NodeId) = source.Id
				*(&sourceDetails[k].InUdpUrl) = source.UdpAddress
				*(&sourceDetails[k].StateUpdateTime) = now
				*(&sourceDetails[k].SourceUrl) = source.Url
			}
			*(&sourceDetails[k].NodeState) = 3
			//如果是外网推流应该存入推流地址
			if len(v.PushUrl) > 0 {
				*(&sourceDetails[k].SourceUrl) = v.PushUrl
			}
		}
		*(&sourceDetails[k].OutUdpUrl) = hub.OutputAddress
		if len(hub.Output.Targets) > 0 {
			*(&sourceDetails[k].AdditionalPublishUrl) = hub.Output.Targets[0].AdditionalPublishUrl
		}

	}
	if len(sourceDetails) > 0 {
		err = tx.Model(&live_schedule.SourceDetail{}).Create(&sourceDetails).Error
		if err != nil {
			e.Log.Errorf("SourceDetail Insert error:%s \r\n", err)
			tx.Rollback()
			return err
		}
	}
	param := make(map[string]interface{})
	//更新任务id和信源状态
	param["out_udp_url"] = hub.OutputAddress
	param["task_id"] = hub.TaskId
	param["source_state"] = 2
	param["update_time"] = now
	if c.IsCustom == 1 {
		param["out_udp_url"] = c.CustomSwitchUrl
	}
	err = tx.Model(&live_schedule.SourceBasic{}).Where("id=?", data.Id).Updates(param).Error
	if err != nil {
		e.Log.Errorf("SourceBasicService Updates error:%s \r\n", err)
		tx.Rollback()
		return err
	}

	c.Id = data.Id
	tx.Commit()
	return nil
}

func (e *RegisterEquipService) GetEquipPage(c *dto.SourceBasicGetPageReq, count *int64) ([]dto.SourceModelRes, error) {
	var err error
	var data live_schedule.SourceBasic
	list := make([]live_schedule.SourceBasic, 0, 5)
	res := make([]dto.SourceModelRes, 0, 5)

	c.IsDelete = 2

	db := e.Orm.Model(&data).
		Scopes(
			cDto.MakeCondition(c.GetNeedSearch()),
			cDto.Paginate(c.GetPageSize(), c.GetPageIndex()),
		).Order("create_time desc")

	if len(c.TenantId) > 0 {
		db.Where("tenant_id = ?", c.TenantId)
	}
	db.Where("kind = ?", c.Kind)

	err = db.Find(&list).Limit(-1).Offset(-1).
		Count(count).Error
	if err != nil {
		e.Log.Errorf("SourceBasicService GetPage error:%s \r\n", err)
		return nil, err
	}

	basicMap := make(map[int]live_schedule.SourceBasic)
	sourceIds := make([]int, 0, 5)
	if len(c.SourceIds) <= 0 {
		for _, v := range list {
			sourceIds = append(sourceIds, v.Id)
			basicMap[v.Id] = v
		}
	} else {
		sourceIds = c.SourceIds
	}
	if len(sourceIds) <= 0 {
		return nil, nil
	}
	//获取信源详情
	details := make([]live_schedule.SourceDetail, 0, 5)
	err = e.Orm.Model(&live_schedule.SourceDetail{}).Where("source_id in ? and area !=2", sourceIds).Find(&details).Error
	if err != nil {
		e.Log.Errorf("get SourceDetail error:%s \r\n", err)
		return nil, err
	}
	for _, v := range list {
		v := v
		r := dto.SourceModelRes{}
		r.Basic = &v
		for _, d := range details {
			if v.Id != d.SourceId {
				continue
			}
			detailModel := live_schedule.SourceDetailModel{}
			detailModel.SourceDetail = d
			if len(d.PushUrl) > 0 {
				arr := strings.Split(d.PushUrl, models.Delimiter)

				for _, url := range arr {
					if strings.Contains(url, "rtmp") {
						detailModel.SourceDetail.PushUrl = url
					}
					if strings.Contains(url, "srt") {
						detailModel.SourceDetail.SrtUrl = url
					}
				}
			}
			r.Details = append(r.Details, detailModel)
		}
		res = append(res, r)
	}
	return res, nil
}

func (e *RegisterEquipService) UpdateEquip(c *dto.EquipUpdateReq) error {

	param := make(map[string]interface{})
	param["source_name"] = c.SourceName
	param["equip_no"] = c.EquipNo
	err := e.Orm.Model(&live_schedule.SourceBasic{}).Where("id=?", c.Id).Updates(&param).Error
	if err != nil {
		e.Log.Errorf("UpdateEquip error:%s \r\n", err)
		return err
	}
	return nil
}

func (e *RegisterEquipService) GetEquipByCode(c *dto.GetEquipByCode) (live_schedule.SourceBasic, error) {

	var basic live_schedule.SourceBasic
	err := e.Orm.Model(&live_schedule.SourceBasic{}).Where("equip_no=?", c.EquipCode).Find(&basic).Error
	if err != nil {
		e.Log.Errorf("get SourceBasic error:%s \r\n", err)
		return basic, err
	}
	details := make([]live_schedule.SourceDetail, 0, 3)
	err = e.Orm.Model(&live_schedule.SourceDetail{}).Where("source_id=?", basic.Id).Find(&details).Error
	if err != nil {
		e.Log.Errorf("get SourceBasic error:%s \r\n", err)
		return basic, err
	}

	for k, detail := range details {
		if len(detail.PushUrl) > 0 {
			arr := strings.Split(detail.PushUrl, models.Delimiter)
			for _, url := range arr {
				if strings.Contains(url, "rtmp") {
					*(&details[k].PushUrl) = url
				}
				if strings.Contains(url, "srt") {
					*(&details[k].SrtUrl) = url
				}
			}
		}
	}

	basic.Details = details

	//获取设备的下车点
	off := live_schedule.TrafficOff{}
	err = e.Orm.Model(&live_schedule.TrafficOff{}).Where("equip_code=?", c.EquipCode).Find(&off).Error
	if err != nil {
		e.Log.Errorf("get TrafficOff error:%s \r\n", err)
		return basic, err
	}
	channelTask := live_schedule.ChannelTask{}
	err = e.Orm.Model(&live_schedule.ChannelTask{}).Where("traffic_off_id=? and task_id!=''", off.Id).Find(&channelTask).Error
	if err != nil {
		e.Log.Errorf("get ChannelTask error:%s \r\n", err)
		return basic, err
	}
	basic.PullUrls = make([]string, 0, 3)
	if channelTask.Id > 0 {
		basic.PullUrls = append(basic.PullUrls, channelTask.OutUrl)
		return basic, err
	}

	cortexTask := live_schedule.CortexTask{}
	err = e.Orm.Model(&live_schedule.CortexTask{}).Where("traffic_off_id=? and task_id!=''", off.Id).Find(&cortexTask).Error
	if err != nil {
		e.Log.Errorf("get CortexTask error:%s \r\n", err)
		return basic, err
	}
	if cortexTask.Id > 0 {
		basic.PullUrls = append(basic.PullUrls, cortexTask.OutUrl)
		return basic, err
	}
	return basic, nil
}

func (e RegisterEquipService) EquipCallBack(d *dto.EquipCallBack) error {

	sourceBasic := live_schedule.SourceBasic{}
	err := e.Orm.Model(&live_schedule.SourceBasic{}).Where("equip_no=?", d.EquipCode).Find(&sourceBasic).Error
	if err != nil {
		e.Log.Errorf("get SourceBasic error:%s \r\n", err)
		return err
	}

	err = e.Orm.Model(&live_schedule.SourceBasic{}).Where("equip_no=?", d.EquipCode).Update("equip_state", d.EquipState).Error
	if err != nil {
		e.Log.Errorf("update SourceBasic error:%s \r\n", err)
		return err
	}
	return nil
}

func (e RegisterEquipService) EquipBeat(d *dto.EquipBeatReq) error {
	now := time.Now().Unix()
	models.EventMap[d.EquipCode] = now
	return nil
}

func (e RegisterEquipService) TimeStopEquip() {

	if len(models.EventMap) <= 0 {
		return
	}

	now := time.Now().Unix()

	offEquipNos := make([]string, 0, 3)
	upEquipNos := make([]string, 0, 3)
	stopEquipNos := make([]string, 0, 3)
	//此设备长时间没收到信号停止任务
	for event, t := range models.EventMap {
		event := event
		if t+30 < now {
			offEquipNos = append(offEquipNos, event)
		} else {
			upEquipNos = append(upEquipNos, event)
		}

		if t+7200 < now {
			stopEquipNos = append(stopEquipNos, event)
		}
	}

	if len(offEquipNos) > 0 {
		err := e.Orm.Model(&live_schedule.SourceBasic{}).Where("equip_no in ? and equip_state=1", offEquipNos).Update("equip_state", 2).Error
		if err != nil {
			e.Log.Errorf("update SourceBasic error:%s \r\n", err)
		}

	}

	if len(upEquipNos) > 0 {
		err := e.Orm.Model(&live_schedule.SourceBasic{}).Where("equip_no in ? and equip_state=2", upEquipNos).Update("equip_state", 1).Error
		if err != nil {
			e.Log.Errorf("update SourceBasic error:%s \r\n", err)
		}
	}

	if len(stopEquipNos) > 0 {
		basics := make([]live_schedule.SourceBasic, 0, 5)
		err := e.Orm.Model(&live_schedule.SourceBasic{}).Where("equip_no in ?", stopEquipNos).Find(&basics).Error
		if err != nil {
			e.Log.Errorf("update SourceBasic error:%s \r\n", err)
		}
		sourceBasicService := SourceBasicService{}
		sourceBasicService.Log = e.Log
		sourceBasicService.Orm = e.Orm
		for _, basic := range basics {
			err = sourceBasicService.StopSource(&dto.StopSourceReq{Id: basic.Id, TenantId: basic.TenantId})
			if err != nil {
				e.Log.Errorf("update SourceBasic error:%s \r\n", err)
			}
		}
	}

	return
}
